package com.example.hbase.service;

import com.example.common.service.TestService;
import com.example.hbase.StudentMapper;
import com.example.hbase.StudentReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class HBaseTestService implements TestService {
    @Override
    public Object doTest(Object param) {
        test();
        return null;
    }

    private void test(){
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "shuxue");
        String[] ioArgs = new String[ ] { "avg_height" };
        String[] otherArgs;
        try {
            otherArgs = new GenericOptionsParser(conf, ioArgs). getRemainingArgs();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        // 一个参数：输出目录
        assert otherArgs.length == 1;
        final FileSystem fileSystem;
        try {
            fileSystem = FileSystem.get(conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        // 删除输出目录，否则随后自动创建时出错
        try {
            fileSystem.delete(new Path(otherArgs[0]), true);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        Job job;
        try {
            job = new Job(conf, "StudentTableMRTest");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        job.setJarByClass(Object.class);
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("hight"));
        scan.setCaching(1024);
        scan.setCacheBlocks(false);
        try {
            TableMapReduceUtil.initTableMapperJob("student", scan,
                    StudentMapper.class,   // mapper 类
                    Text.class,            // mapper输出key
                    IntWritable.class,      // mapper输出value
                    job);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        job.setReducerClass(StudentReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //job.setOutputFormatClass(TextOutputFormat.class);

        job.setNumReduceTasks(1);
        //Path outputPath = new Path("file:///user/data/");
        Path outputPath = new Path("hdfs://192.168.0.130:9000/output");
        try {
            if(fileSystem.exists(outputPath)) {
                fileSystem.delete(outputPath,true);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        FileOutputFormat.setOutputPath(job, outputPath);
        try {
            if (!job.waitForCompletion(true)) {
                throw new IOException("Error with job!");
            }
        } catch (IOException | InterruptedException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
}
